package udf.evaluation;

import java.io.IOException;

import java.io.UnsupportedEncodingException;

import java.util.LinkedList;

import org.apache.http.HttpException;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.data.DataType;

public class fetchGoogleSuggestions extends EvalFunc<DataBag> {

	BagFactory mBagFactory = BagFactory.getInstance();
	String AGENT_LIST = "/home/sergio/projects/dictionary/user_agent_list.txt";

	// static ArrayList<String> mTables = new ArrayList<String>();

	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";
	
	
	public LinkedList<String> getGoogleSuggestions(String query, String path,
			long delay) {
		LinkedList<String> suggestions = null;
		try {
			
		
			suggestions = util.webSearch.GoogleQuery.getRelatedSearchesUDF(query,
					path, delay);
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (HttpException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		return suggestions;
	}

	public DataBag exec(Tuple input) throws IOException {

		if (input.get(0) == null) {
			return null;
		}
		
		else {

			try {
				DataBag output = mBagFactory.newDefaultBag();

				String query = (String) input.get(0);
				String path = (String) input.get(1);
				Integer delay = (Integer) input.get(2);
				LinkedList<String> list = getGoogleSuggestions(query.trim(),
						path, delay);

				for (int i = 0; i < list.size(); i++) {

					String expansion = list.get(i);
					TupleFactory mTupleFactory = TupleFactory.getInstance();
					Tuple tuple = mTupleFactory.newTuple();
					tuple.append(query);
					tuple.append(expansion);
					tuple.append(i + 1);
					System.out.println(query + "\t" + expansion + "\t" + (i+1));
					// Object tuple = mTupleFactory.newTuple(expansion);

					output.add(tuple);
				}

				// StringTokenizer tok = new StringTokenizer((String)o,
				// " \",()*", false);
				// while (tok.hasMoreTokens())
				// output.add(mTupleFactory.newTuple(tok.nextToken()));
				// return output;

				return output;

			} catch (Exception e) {
				System.err.println("Failed to process input; error - "
						+ e.getMessage());
				return null;
			}

		}

	}

	public Schema outputSchema(Schema input) {
		try {
			Schema bagSchema = new Schema();
			
			bagSchema.add(new Schema.FieldSchema("query",
					DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("suggestion",
					DataType.CHARARRAY));
			bagSchema.add(new Schema.FieldSchema("rank", DataType.INTEGER));

			return new Schema(new Schema.FieldSchema(getSchemaName(this
					.getClass().getName().toLowerCase(), input), bagSchema,
					DataType.BAG));

		} catch (Exception e) {
			return null;
		}
	}

}